package com.bfxy.disruptor.common.event;

import java.util.Arrays;

import com.bfxy.disruptor.common.ConsumerRepository;
import com.bfxy.disruptor.common.Disruptor;
import com.bfxy.disruptor.common.WorkHandler;
import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.consume.ConsumeFactory;
import com.bfxy.disruptor.common.consume.EventConsumer;

public class EventHandlerGroup<T> {
	private final Disruptor<T> disruptor;
	private final ConsumerRepository<T> consumerRepository;
	private final Sequence[] sequences;

	public EventHandlerGroup(final Disruptor<T> disruptor, final ConsumerRepository<T> consumerRepository,
			final Sequence[] sequences) {
		this.disruptor = disruptor;
		this.consumerRepository = consumerRepository;
		this.sequences = Arrays.copyOf(sequences, sequences.length);
	}

	public EventHandlerGroup<T> and(final EventHandlerGroup<T> otherHandlerGroup) {
		final Sequence[] combinedSequences = new Sequence[this.sequences.length + otherHandlerGroup.sequences.length];
		System.arraycopy(this.sequences, 0, combinedSequences, 0, this.sequences.length);
		System.arraycopy(otherHandlerGroup.sequences, 0, combinedSequences, this.sequences.length,
				otherHandlerGroup.sequences.length);
		return new EventHandlerGroup<T>(disruptor, consumerRepository, combinedSequences);
	}

	public EventHandlerGroup<T> and(final EventConsumer... processors) {
		Sequence[] combinedSequences = new Sequence[sequences.length + processors.length];

		for (int i = 0; i < processors.length; i++) {
			consumerRepository.add(processors[i]);
			combinedSequences[i] = processors[i].getSequence();
		}
		System.arraycopy(sequences, 0, combinedSequences, processors.length, sequences.length);

		return new EventHandlerGroup<T>(disruptor, consumerRepository, combinedSequences);
	}

	public EventHandlerGroup<T> then(final EventHandler<? super T>... handlers) {
		return handleEventsWith(handlers);
	}

	public EventHandlerGroup<T> then(final ConsumeFactory<T>... eventProcessorFactories) {
		return handleEventsWith(eventProcessorFactories);
	}

	public EventHandlerGroup<T> thenHandleEventsWithWorkerPool(final WorkHandler<? super T>... handlers) {
		return handleEventsWithWorkerPool(handlers);
	}

	public EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers) {
		return disruptor.createEventProcessors(sequences, handlers);
	}

	public EventHandlerGroup<T> handleEventsWith(final ConsumeFactory<T>... eventProcessorFactories) {
		return disruptor.createEventProcessors(sequences, eventProcessorFactories);
	}

	public EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<? super T>... handlers) {
		return disruptor.createWorkerPool(sequences, handlers);
	}

	public SequenceBarrier asSequenceBarrier() {
		return disruptor.getRingBuffer().newBarrier(sequences);
	}
}
